//****************************************************************************
//File: Master.java
//This Java Source File was implemented in the same way as Prof. Alan Kaminsky
//did in the Publisher.java and Subscriber.java class of his lecture notes.
//URL: http://www.cs.rit.edu/~ark/730/jms/jms.shtml
//The Master programs subscribe to the result topic of matrix multiplication
//and generates input matrices and publishes to the matrix multiplication
//topic.
//***************************************************************************
import java.io.Serializable;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Class Master provides an object for generating input matrices
 * and publishing it to the slaves and subscribing for the result
 * topic and generating the end to end response time.
 *
 * @author  Vivek Narang (vsn6751@rit.edu)
 * @author  Alan Kaminsky
 * @version 05-May-2012
 */
public class Master {
// Data Members
	// Default user and password for the MQ Connection object
	private static String user = ActiveMQConnection.DEFAULT_USER;
	private static String password = ActiveMQConnection.DEFAULT_PASSWORD;
	// URL at which the connection is made
	private static String url;
	// Topic where input matrices are published
	private static String inputTopic = "MatrixMultiplication";
	// Topic where resultant column is published
	private static String resultTopic = "Result";
	// Keep a count of the number of columns
	private static int colCount = 0;
	// Message Consumer
	private static MessageConsumer consumer;
	// Connection object
	private static Connection connection;
	// Resultant matrix
	private static int[][] res;
	// Session object
	private static Session session;
	// Start time when matrices are sent
	private long startTime;
	// End time when all columns are received
	private long endTime;
	// Number of slaves used to perform multiplication
	private int noOfSlaves;
	
	/**
	 * Create the matrix input object and publish it
	 * to the input topic and receive all the resultant
	 * columns by various slaves. Also record the end to end
	 * response time from sending the matrix to receiving the
	 * result back.
	 *
	 * @param  size  The size of the square matrix.
	 * @param  noOfNodes  The number of slaves used to generate
	 * 					  the resultant matrix.
	 * 
	 * @exception  Exception
	 *     Thrown if there was a problem with execution of the method.
	 *
	 */
	public void performMultiplication(int size, int noOfNodes,
			String host, int port)
		throws Exception {
		
		// Initialize the resultant matrix
		res = new int[size][size];
		
		// Initialize the number of slaves
		noOfSlaves = noOfNodes;
		
		// Create connection
		url = "failover:tcp://" + host + ":" + port;
		ActiveMQConnectionFactory connectionFactory =
			new ActiveMQConnectionFactory (user, password, url);
		connection = connectionFactory.createConnection();
		connection.setExceptionListener (new ExceptionListener() {
			public void onException (JMSException exc) {
				exc.printStackTrace (System.err);
				System.exit (1);
			}
		});
		connection.start();
		
		// Create session
		session = connection.createSession(false,
				Session.AUTO_ACKNOWLEDGE);
		
		// Create a destination for publishing the matrices to slaves
		Destination destinationPublish = session.createTopic (inputTopic);
		
		// Create a destination for receiving the result in result queue
		Destination destinationReceiver = session.createTopic (resultTopic);
		
		// Create  producer
		MessageProducer producer = session.createProducer (destinationPublish);
		
		// Create consumer
		consumer = session.createConsumer (destinationReceiver);
		
		
		//synchronized(this) {
			// Receive the resultant column as a ObjectMessage
			consumer.setMessageListener (new MessageListener() {
				public void onMessage (Message msg) {
					if (msg instanceof ObjectMessage) {
						try {
							// Receive the result object
							ResultantColumn result = (ResultantColumn)
								((ObjectMessage)msg).getObject();
							
							// Increment the count
							colCount++;
							
							// Get the result column and the column ID
							// to put it in the final resultant matrix
							int[] resultColumn = result.getResultCol();
							int resultColId = result.getColumnId();
							
							for(int i = 0; i < resultColumn.length; i++) {
								res[i][resultColId] = resultColumn[i];
							}
							
							// If all the columns have been received
							if(resultColumn.length == colCount) {
								// Set the counter to 0
								colCount = 0;
								
								// Record the end time
								endTime = System.currentTimeMillis();
								System.out.printf("%d %d %d\n",
										resultColumn.length,
										noOfSlaves,
										(endTime - startTime));
								
								// Close the connection
								connection.close();
							}
						}
						catch (JMSException exc) {
							System.out.printf ("Could not get message text%n");
						}
					}
					else {
						System.out.printf ("Unknown message type%n");
					}
				}
			});
		//}
				
		// Generate the matrix input object
		MatrixInput matrix = new MatrixInput(size, size, noOfNodes);
		
		// Create a Object Message
		ObjectMessage objMsg = session.createObjectMessage();
		
		// Start the timer
		startTime = System.currentTimeMillis();
		objMsg.setObject((Serializable) matrix);
		// Send the object to the slaves
		producer.send (objMsg);
	}
}